package com.cfs.ipc;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.cfs.io.DataOutputBuffer;
import com.cfs.utils.IOUtils;


public class Listener implements Runnable{
	
	private static Log LOG = LogFactory.getLog(Listener.class);
	
	private ServerSocketChannel serverSocketChannel;
	private SocketChannel socketChannel;
	private Selector selector;
	private ByteBuffer receiveBuffer = null;
	private ByteBuffer sendBuffer = null;
	private ByteBuffer header = ByteBuffer.allocate(4);

	private Server server;
	
	public Listener(Server server) {
		this.server = server;
		init();
	}

	private void init() {
		try {
			serverSocketChannel = ServerSocketChannel.open();
			serverSocketChannel.socket().bind(new InetSocketAddress(server.getPort()));
			serverSocketChannel.configureBlocking(false);
			selector = Selector.open();
			serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
			LOG.info("listener port: " + server.getPort());
		} catch (IOException e) {
			e.printStackTrace();
		}		
	}

	@Override
	public void run() {
		try {
			while(server.isRunning()) {
				if (selector.select() > 0) {
					Set<SelectionKey> selectionKeys = selector.selectedKeys();
					Iterator<SelectionKey> iterator = selectionKeys.iterator();
					while (iterator.hasNext()) {
						SelectionKey selectionKey = iterator.next();
						handleKey(selectionKey);
						iterator.remove();
					}
				}
			}
		}catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void handleKey(SelectionKey selectionKey) throws Exception {
		int count;
		try{
			if (selectionKey.isAcceptable()) {
				serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
				socketChannel = serverSocketChannel.accept();
				socketChannel.configureBlocking(false);
				socketChannel.register(selector, SelectionKey.OP_READ);
			} else if (selectionKey.isReadable()) {
				socketChannel = (SocketChannel) selectionKey.channel();

				header.clear();
				IOUtils.channelIO(socketChannel, null, header);
				header.flip();
				int length = header.getInt();
				
				receiveBuffer = ByteBuffer.allocate(length);
				count = IOUtils.channelIO(socketChannel, null, receiveBuffer);
				receiveBuffer.flip();
				LOG.info("2.listener receiveBuffer length: " + receiveBuffer.capacity());
				if (count > 0) {
					Invocation invocation = new Invocation();
					DataInputStream in = new DataInputStream(new ByteArrayInputStream(receiveBuffer.array()));
					invocation.readFields(in);
					server.call(invocation);
					SelectionKey key = socketChannel.register(selector, SelectionKey.OP_WRITE);
					key.attach(invocation);
				}
			} else if (selectionKey.isWritable()) {
				socketChannel = (SocketChannel) selectionKey.channel();
				
				Invocation invocation = (Invocation) selectionKey.attachment();
				DataOutputBuffer buffer = new DataOutputBuffer();
				invocation.write(buffer);
				
				int bufLen = buffer.getLength();
				ByteArrayOutputStream arrayOut = new ByteArrayOutputStream();
				DataOutputStream dataOut = new DataOutputStream(arrayOut);
				dataOut.writeInt(bufLen);
				dataOut.write(buffer.getData(), 0 ,bufLen);
				
				sendBuffer = ByteBuffer.wrap(arrayOut.toByteArray());
				IOUtils.channelIO(null, socketChannel, sendBuffer);
				LOG.info("3.listener sendBuffer length: " + sendBuffer.capacity());
				socketChannel.register(selector, SelectionKey.OP_READ);
			}
		}catch (Exception e) {
			selectionKey.cancel();
			if (selectionKey.channel() != null){
				try {
					selectionKey.channel().close();
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			}
		}
		
	}

}
